package com.miao.im.dispatcher;

import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.miao.im.common.Message;
import com.miao.im.common.enums.MessageType;
import com.miao.im.dto.SocketMessageDto;
import com.miao.im.server.session.SessionContext;
import com.miao.im.support.handler.AbstractMessageHandler;
import com.miao.im.support.handler.MessageHandlerFactory;
import com.rabbitmq.client.Channel;
import io.netty.util.ReferenceCountUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Mono;

import java.io.IOException;
import java.nio.charset.StandardCharsets;

/**
 * @Auther: miaoguoxin
 * @Date: 2021/2/24 18:09
 * @Description:
 */
@Slf4j
@Component
public class MessageDispatcher {
    @Autowired
    private MessageHandlerFactory messageHandlerFactory;

    /**
    * 消息入口
    * @author miaoguoxin
    * @date 2021/5/20
    */
    public Mono<Void> push(Message message, SessionContext sessionContext) {
        AbstractMessageHandler<?> handler = messageHandlerFactory.createHandler(message.getCommandType());
        return handler.dispatch(message, sessionContext);
    }


    /**
     * 单通道/广播读取消息
     * @author miaoguoxin
     * @date 2021/3/16
     */
    @RabbitListener(queues = {"#{systemProperties['singleQueue']}", "#{systemProperties['broadcastQueue']}"}, ackMode = "MANUAL")
    public void consumeSingleMessage(org.springframework.amqp.core.Message message, Channel channel) throws Exception {
        String data = new String(message.getBody(), StandardCharsets.UTF_8);
        if (StrUtil.isBlank(data)) {
            rejectMsg(channel, message.getMessageProperties().getDeliveryTag());
            return;
        }
      //  log.info("接收到消费消息:{}",data);
        SocketMessageDto messageDto;
        try {
            messageDto = JSON.parseObject(data, SocketMessageDto.class);
        } catch (Exception e) {
            log.error("转换消息json格式异常", e);
            rejectMsg(channel, message.getMessageProperties().getDeliveryTag());
            return;
        }
        Message bufMessage = convertBufMessage(messageDto);
        try {
            AbstractMessageHandler<?> handler = messageHandlerFactory.createHandler(bufMessage.getCommandType());
            handler.read(bufMessage);
            ackMsg(channel, message.getMessageProperties().getDeliveryTag());
        } catch (Exception e) {
            log.error("消费消息异常:", e);
            Boolean redelivered = message.getMessageProperties().getRedelivered();
            if (redelivered) {
                // 消息已重复处理失败, 扔掉消息
                rejectMsg(channel, message.getMessageProperties().getDeliveryTag());
            } else {
                nackMsg(channel, message.getMessageProperties().getDeliveryTag());
            }
        }
    }

    /**
     * @author miaoguoxin
     * @date 2020/9/16
     */
    private static void nackMsg(Channel channel, long deliveryTag) {
        try {
            channel.basicNack(deliveryTag, false, true);
        } catch (IOException e) {
            log.error("nack message failed", e);
        }
    }


    /**
     * @author miaoguoxin
     * @date 2020/9/16
     */
    private static void rejectMsg(Channel channel, long deliveryTag) {
        try {
            channel.basicReject(deliveryTag, false);
        } catch (IOException e) {
            log.error("reject message failed", e);
        }
    }

    /**
     * @author miaoguoxin
     * @date 2020/9/16
     */
    private static void ackMsg(Channel channel, long deliveryTag) {
        try {
            channel.basicAck(deliveryTag, false);
        } catch (IOException e) {
            log.error("ack message failed", e);
        }
    }

    private static Message convertBufMessage(SocketMessageDto bizMsg) {
        Message bufMessage;
        if (bizMsg.getMessageType() == MessageType.REQ.getCode()) {
            bufMessage = Message.writeRequest(bizMsg.getVersion(), bizMsg.getSequence(), bizMsg.getCommandType(), bizMsg.getBody());
        } else {
            bufMessage = Message.writeResponse(bizMsg.getVersion(), bizMsg.getSequence(), bizMsg.getCommandType(), bizMsg.getBody());
        }
        return bufMessage;
    }
}
